﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using NetMicro.Core.Extensions;
using NetMicro.EventBus.Diagnostics;
using NetMicro.EventBus.Enums;
using NetMicro.EventBus.Extensions;
using NetMicro.EventBus.Messages;
using NetMicro.Trace.Propagator;

namespace NetMicro.EventBus.RabbitMQ
{
    public class RabbitMQEventSubscriber : IEventSubscriber
    {
        private static readonly ActivitySource RabbitMQEventSubscriberActivitySource = new ActivitySource(nameof(RabbitMQEventSubscriber));
        private static readonly TraceContextPropagator RabbitMQEventSubscriberPropagator = new TraceContextPropagator();
        public RabbitMQEventSubscriber(IOptionsMonitor<EventBusRabbitMQOptions> options,
            IChannelFactory channelFactory,
            ILogger<RabbitMQEventSubscriber> logger,
            IMessageSerializer messageSerializer,
            IMessageListener messageListener,
            IEventErrorHandler eventErrorHandler)
        {
            Options = options;
            ChannelFactory = channelFactory;
            Logger = logger;
            MessageSerializer = messageSerializer;
            MessageListener = messageListener;
            EventErrorHandler = eventErrorHandler;
            _consumers = new ConcurrentDictionary<string, EventingBasicConsumer>();
        }
        private readonly ConcurrentDictionary<string, EventingBasicConsumer> _consumers;
        private IOptionsMonitor<EventBusRabbitMQOptions> Options { get; }
        private ILogger<RabbitMQEventSubscriber> Logger { get; }
        private IMessageSerializer MessageSerializer { get; }
        private IMessageListener MessageListener { get; }
        private IChannelFactory ChannelFactory { get; }

        private IEventErrorHandler EventErrorHandler { get; }
        /// <summary>
        /// 
        /// </summary>
        /// <param name="eventHandlerDescriptor"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task SubscribeAsync(EventHandlerDescriptor eventHandlerDescriptor, CancellationToken cancellationToken)
        {

            string eventName = eventHandlerDescriptor.EventName;
            string eventHandlerName = eventHandlerDescriptor.EventHandlerName;

            var Channel = await ChannelFactory.CreateChannelAsync();

            // 注册基础通信交换机,类型topic
            Channel.ExchangeDeclare(Options.CurrentValue.Exchange, "topic", true);
            // 定义队列
            Channel.QueueDeclare(eventHandlerName, true, false, false);
            // 绑定队列到交换机以及routing key
            Channel.QueueBind(eventHandlerName, Options.CurrentValue.Exchange,
                eventName);

            EventingBasicConsumer consumer = _consumers.GetOrAdd(eventHandlerName, r => new EventingBasicConsumer(Channel));
            consumer.Received += async (sender, e) =>
            {
                if (cancellationToken.IsCancellationRequested)
                    return;

                MessageTransfer message;
                try
                {
                    message = MessageSerializer.DeserializeE<MessageTransfer>(e.Body.ToArray());
                }
                catch (Exception exception)
                {
                    Logger.LogError(exception, "[EventBus-RabbitMQ] deserialize message from rabbit error.");
                    return;
                }

                if (message is null)
                {
                    Logger.LogError("[EventBus-RabbitMQ] deserialize message from rabbit error.");
                    return;
                }

                if (message.EventName != eventName)
                {
                    Logger.LogError(
                        $"[EventBus-RabbitMQ] received invalid event name \"{message.EventName}\", expect \"{eventName}\".");
                    return;
                }
                var parentContexttwo = RabbitMQEventSubscriberPropagator.Extract(default, message.Items, TraceContext.ExtractTraceContextFromBasicProperties);
                using (var activity = RabbitMQEventSubscriberActivitySource.StartActivity(message.EventHandlerName, ActivityKind.Consumer, parentContexttwo))
                {

                    Logger.LogDebug(
                        $"[EventBus-RabbitMQ] received msg: {message.ToJson()}.");

                    // 处理消息
                    message.EventHandlerName = eventHandlerName;
                    MessageReceiveResult res = await MessageListener.OnReceiveAsync(message, cancellationToken).ConfigureAwait(false);
                    if (res.MessageReceive == MessageReceive.Failed)
                    {
                        message.Exception = res.ExceptionMsg;
                        await EventErrorHandler.OnErrorAsync(message, cancellationToken).ConfigureAwait(false);
                    }
                    // 一定要在消息接收处理完成后才确认ack
                    Channel.BasicAck(e.DeliveryTag, false);
                }

            };

            consumer.Registered += (sender, e) =>
            {
                Logger.LogInformation(
                    $"[EventBus-RabbitMQ] event handler \"{eventHandlerName}\" has been registered.");
            };
            consumer.Shutdown += (sender, e) =>
            {
                Logger.LogWarning(
                    $"[EventBus-RabbitMQ] event handler \"{eventHandlerName}\" has been shutdown.");
            };
            Channel.BasicConsume(eventHandlerName, false, consumer);


        }

        private IEnumerable<string> ExtractTraceContextFromBasicProperties(Dictionary<string, string> props, string key)
        {
            try
            {
                if (props.TryGetValue(key, out var value))
                {

                    List<string> list = new List<string>();
                    list.Add(value.ToString());
                    return list.ToArray();
                }
            }
            catch (Exception ex)
            {

            }

            return Enumerable.Empty<string>();
        }
    }
}